package com.chat.kafka.message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Auther created by haoteng on 2019/08/15
 */
@Component
public class KafkaMessage {
    Logger logger = LoggerFactory.getLogger(KafkaMessage.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void send() {

        for (int i = 0; i < 100; i++) {
            String data = i + "----------" + System.currentTimeMillis();
            ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(Topics.KAFKA_TOPIC_DEMO, data);
            listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    logger.info("kafka sendMessage error, ex = {}, topic = {}, data = {}", throwable, Topics.KAFKA_TOPIC_DEMO, data);
                }

                @Override
                public void onSuccess(SendResult<String, String> stringStringSendResult) {
                    logger.info("kafka sendMessage success topic = {}, data = {}", Topics.KAFKA_TOPIC_DEMO, data);
                }
            });
        }


    }

    @KafkaListener(topics = Topics.KAFKA_TOPIC_DEMO)
    public void listener(String message) {
        logger.info("consumer accept message from product :" + message);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
